/**
 * DVN: Distributed Virtual Network Simulation Platform
 * Copyright (c) 2006-2009 Tyson Malchow <tyson.malchow@gmail.com>
 *
 * This file is part of the DVN distributed virtual network simulation
 * program.
 *
 * DVN is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * DVN is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with DVN.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <stdio.h>
#include <string.h>

#include "global.h"
#include "sysconfig.h"
#include "sim.h"
#include "events.h"
#include "netconst.h"
#include "routing.h"
#include "simop.h"
#include "timer.h"
#include "utility.h"
#include "context.h"
#include "list.h"

sim_worker_tp sim_worker_create (pipecloud_tp pipecloud, int slave_id, int process_id, unsigned int num_slaves, unsigned int num_workers, unsigned int max_wrkrs_per_slave) {
	sim_worker_tp rv;

	rv=malloc(sizeof(*rv));

	rv->pipecloud = pipecloud;
	rv->events = events_create();
	rv->stalled_simops = list_create();
	rv->mod_mgr = module_mgr_create();
	rv->node_tracking = hashtable_create(sysconfig_get_int("sim_nodetrack_hashsize"),sysconfig_get_float("sim_nodetrack_hashgrowth"));
	rv->timer_mgr = dtimer_create_manager(rv->events);
	rv->ascheme = vci_create_addressing_scheme(num_slaves, max_wrkrs_per_slave);
	rv->vci_mgr = vci_mgr_create(rv->events, slave_id, process_id, rv->ascheme);

	rv->num_workers = num_workers;
	rv->process_id = process_id;
	rv->net_max_delay = rv->net_min_delay = 0;
	rv->worker_states = malloc(sizeof(*rv->worker_states) * (num_workers + 1));
	rv->current_time = rv->current_window = PTIME_INVALID;

	for(int i=0; i <= num_workers; i++) {
		rv->worker_states[i].last_completed = PTIME_INVALID;
		rv->worker_states[i].last_processed = PTIME_INVALID;
		rv->worker_states[i].next_event = PTIME_INVALID;
		rv->worker_states[i].valid = 0;
	}

	rv->worker_states[process_id].valid = 1;
	rv->mode = sim_worker_mode_spool;

	return rv;
}

void sim_worker_deposit(sim_worker_tp worker, int frametype, nbdf_tp frame) {
	switch(frametype) {
		case SIM_FRAME_START: {
			nbdf_read(frame, "tt", &worker->net_max_delay, &worker->net_min_delay);
			debugf("SWorker (%d): Got frame start, max delay: %llu, min delay: %llu\n", worker->process_id, worker->net_max_delay, worker->net_min_delay);
			worker->mode = sim_worker_mode_simulating;
			break;
		}

		case SIM_FRAME_OP: {
			simop_tp sop = simop_nbdf_decode(frame);
			if(sop)
				events_schedule(worker->events, sop->target_time, sop, EVENTS_TYPE_SIMOP);
			break;
		}

		case SIM_FRAME_TRACKNODE: {
			unsigned int tracking_id;
			in_addr_t addr;
			sim_worker_nodetracker_tp nt;

			nbdf_read(frame, "ia", &tracking_id, &addr);
			debugf("SWorker (%d): Got tracking response for ID %d - ip: %s\n", worker->process_id, tracking_id, inet_ntoa_t(addr));

			nt = hashtable_get(worker->node_tracking, tracking_id);
			if(!nt) {
				nt = malloc(sizeof(*nt));
				if(!nt)
					printfault(EXIT_NOMEM,"sim_worker_deposit: Out of memory");
				nt->track_id = tracking_id;
				hashtable_set(worker->node_tracking, tracking_id, nt);
			}

			nt->addr = addr;
			nt->valid = 1;

			break;
		}

		case SIM_FRAME_VCIDELIV: {
			vci_deposit(worker->vci_mgr, frame);
			break;
		}

		case SIM_FRAME_STATE: {
			unsigned int source_worker;
			ptime_t completed, processed, next;

			nbdf_read(frame, "ittt", &source_worker, &completed, &processed, &next);

			worker->worker_states[source_worker].last_completed = completed;
			worker->worker_states[source_worker].last_processed = processed;
			worker->worker_states[source_worker].next_event = next;
			worker->worker_states[source_worker].valid = 1;

			//debugf("SWorker (%d): Got state from %d: @%llu @%llu @%llu\n", worker->process_id, source_worker, completed, processed, next);
		}
	}
}

int sim_worker_timecalc(sim_worker_tp worker, ptime_t next_event_time, ptime_t last_event_time) {
	ptime_t min_completed = PTIME_MAX;
	ptime_t max_processed = PTIME_INVALID;
	ptime_t min_next      = PTIME_MAX;
	ptime_t run_to;
	unsigned int i, state_change = 0;

	if(list_get_size(worker->stalled_simops) > 0) {
		debugf("Stalled for simop wait!\n");
		return 0;
	}

	debugf("SWorker (%d): Timecalc. My state: (@%llu @%llu @%llu)\n",worker->process_id,
                        worker->worker_states[worker->process_id].last_completed,
                        worker->worker_states[worker->process_id].last_processed,
                        worker->worker_states[worker->process_id].next_event); 

	/* update our own state values, when appropriate */
	if(next_event_time != PTIME_INVALID && next_event_time != worker->worker_states[worker->process_id].next_event) {
		worker->worker_states[worker->process_id].next_event = next_event_time;
		state_change = 1;
	}

	if(last_event_time != PTIME_INVALID && last_event_time != worker->worker_states[worker->process_id].last_processed) {
		worker->worker_states[worker->process_id].last_processed = last_event_time;
		state_change = 1;
	}

	/* this function will be called whenever the current time window is "done". so, update our last_completed...*/
	if(worker->worker_states[worker->process_id].last_completed != worker->current_window) {
		worker->worker_states[worker->process_id].last_completed = worker->current_window;
		state_change = 1;
	}

	if(state_change) {
		/* report our new state to the other folks */
		nbdf_tp state_frame = nbdf_construct("ittt", worker->process_id,
				worker->worker_states[worker->process_id].last_completed,
				worker->worker_states[worker->process_id].last_processed,
				worker->worker_states[worker->process_id].next_event
		);
		debugf("SWorker (%d):   New state, broadcast (@%llu @%llu @%llu)\n", 
			worker->process_id,
			worker->worker_states[worker->process_id].last_completed,
			worker->worker_states[worker->process_id].last_processed,
			worker->worker_states[worker->process_id].next_event);

		dvn_packet_route(DVNPACKET_WORKER_BCAST, DVNPACKET_LAYER_SIM, 0, SIM_FRAME_STATE, state_frame);
		nbdf_free(state_frame);

	}

	/* figure out our bounds ! */
	for(i=0; i <= worker->num_workers; i++) {
		if(!worker->worker_states[i].valid)
			break;

		/* find the soonest time reported completed */
		if(worker->worker_states[i].last_completed < min_completed) /* PTIME_INVALID here is okay - means 0 */
			min_completed = worker->worker_states[i].last_completed;

		/* find the soonest next VALID event */
		if(worker->worker_states[i].next_event < min_next && worker->worker_states[i].next_event != PTIME_INVALID)
			min_next = worker->worker_states[i].next_event;

		/* find the latest processed event */
		if(worker->worker_states[i].last_processed > max_processed)
			max_processed = worker->worker_states[i].last_processed;
	}

	if(i <= worker->num_workers) { /*encountered an invalid state frame*/
		debugf("SWorker (%d):    Break, not enough state infomation to proceed.\n", worker->process_id);
		return 0;
	}

	/* if min_completed is invalid, it means we're at the very start of the sim. so, start time is the first event! */
	if(min_completed == PTIME_INVALID) {
		run_to = min_next + worker->net_min_delay - 1;
		debugf("min_completed == PTIME INVALID, %llu = %llu + %llu - 1\n", run_to ,min_next, worker->net_min_delay);

	/* if the last reported completed is greater than the influence possibility of the latest event processed, we can 
	 * safely timejump to the next event. */
	} else if(worker->worker_states[worker->process_id].last_completed >= max_processed + worker->net_max_delay) {
		run_to = min_next + worker->net_min_delay - 1;
		debugf("My LastComp >= (max proc + net max delay), (TJ) %llu = %llu + %llu - 1\n", run_to ,min_next, worker->net_min_delay);


	/* by default, we run out as far as we can per network delay model */
	} else {
		run_to = min_completed + worker->net_min_delay - 1;
		debugf("Case3: %llu = %llu + %llu - 1\n", run_to ,min_completed, worker->net_min_delay);
	}

	debugf("SWorker (%d):    RT: @%llu minComp:@%llu minNext:@%llu maxProc:@%llu\n", worker->process_id, run_to, min_completed, min_next, max_processed);

	if(run_to > worker->current_window) {
		debugf("SWorker (%d):    Time Window Advance: @%llu -> @%llu\n", worker->process_id, worker->current_window, run_to);
		//if(run_to < worker->current_window) 
		//	{ debugf("           REVERSE TIME BAD -------------------------------\n"); }
		worker->current_window = run_to;

		return 1;
	}

	return 0;
}

int sim_worker_setstate(sim_worker_tp worker, ptime_t last_event_time, ptime_t next_event_time) {
	int rv = 0;

	if(list_get_size(worker->stalled_simops) > 0)
		return 0;

	worker->worker_states[worker->process_id].valid = 1;

	if(worker->worker_states[worker->process_id].last_completed != worker->current_window) {
		rv++;
		worker->worker_states[worker->process_id].last_completed = worker->current_window;
	}

	if(worker->worker_states[worker->process_id].last_processed != last_event_time) {
		rv++;
		worker->worker_states[worker->process_id].last_processed = last_event_time;
	}

	if(worker->worker_states[worker->process_id].next_event != next_event_time) {
		rv++;
		worker->worker_states[worker->process_id].next_event = next_event_time;
	}

	if(rv) {
		/* broadcast our new state. */
		nbdf_tp state_frame = nbdf_construct("ittt", worker->process_id,
				worker->worker_states[worker->process_id].last_completed,
				worker->worker_states[worker->process_id].last_processed,
				worker->worker_states[worker->process_id].next_event
		);
		dvn_packet_route(DVNPACKET_WORKER_BCAST, DVNPACKET_LAYER_SIM, 0, SIM_FRAME_STATE, state_frame);
		nbdf_free(state_frame);
		debugf("SWorker (%d): State broadcast @%llu @%llu @%llu\n", 
			worker->process_id,
			worker->worker_states[worker->process_id].last_completed,
			worker->worker_states[worker->process_id].last_processed,
			worker->worker_states[worker->process_id].next_event);
	}

	return rv;
}

int sim_worker_heartbeat(sim_worker_tp worker) {
	unsigned int num_events_processed = 0;
	ptime_t next_event_time = PTIME_INVALID, last_event_time = PTIME_INVALID;

	debugf("SWorker (%d): Heartbeat\n", worker->process_id);

	/* vars for the current event */
	ptime_t event_at;
	int event_type;
	void * event;
	simop_tp simop_event;

	/* we can't do anything before we've gotten the start packet */
	if(worker->mode != sim_worker_mode_simulating)
		return 1;

	/* first, attempt to process any stalled simop events. */
	while((simop_event = list_pop_front(worker->stalled_simops)) != NULL) {
		/* set the current time for SNRI callbacks */
		worker->current_time = simop_event->target_time;

		if(!sim_worker_opexec(worker, simop_event)) {
			list_push_front(worker->stalled_simops, simop_event);
			break;
		}
	}

	next_event_time = events_get_next_time(worker->events);

	while(sim_worker_timecalc(worker, next_event_time, last_event_time)) {
		/* while in bounds, dequeue and process! */
		while(worker->mode == sim_worker_mode_simulating && (next_event_time = events_get_next_time(worker->events)) <= worker->current_window) {
			event = events_dequeue(worker->events, &event_at, &event_type);

			if(!event)
				break;

			/* set the current time for SNRI callbacks */
			worker->current_time = event_at;

			switch(event_type) {
			case EVENTS_TYPE_DTIMER:
				dtimer_exec_event(worker->timer_mgr, event);
				break;

			case EVENTS_TYPE_VCI:
				vci_exec_event(worker->vci_mgr, event);
				break;

			case EVENTS_TYPE_SIMOP:
				if(!sim_worker_opexec(worker, event)) {
					list_push_back(worker->stalled_simops, event);
					continue;
				}
				break;
			}

			num_events_processed++;
			last_event_time = next_event_time;
		}


		debugf(" Event Exec\n");

		if(worker->mode != sim_worker_mode_simulating)
			break;
	}

	if(worker->mode != sim_worker_mode_simulating)
		return 0;

	return 1;
}




int sim_worker_opexec_cnodes(sim_worker_tp wo, simop_tp sop) {
	int i;
	in_addr_t bootstrap_ip = INADDR_NONE;
	module_tp  module = module_get_module(wo->mod_mgr, sop->detail.cnodes.module_id);

	if(!module)
		return 1;

	debugf("SWorker (%d): Spawning %d nodes @%llu.\n",wo->process_id,sop->detail.cnodes.quantity, sop->target_time);

	/* let's find out whether or not we've already recieved IP information for the bootstrap node */
	if(sop->detail.cnodes.flags & SIMOP_CNODES_BOOTSTRAP) {
		sim_worker_nodetracker_tp nodetrack =
			hashtable_get(wo->node_tracking, sop->detail.cnodes.bootstrap_id);
		if(nodetrack && nodetrack->valid) {
			bootstrap_ip = nodetrack->addr;
			debugf("SWorker (%d): Bootstrapping from already valid address %s on tracking ID %d\n", wo->process_id, inet_ntoa_t(nodetrack->addr), nodetrack->track_id);
		} else {
			if(!nodetrack) {
				nodetrack = malloc(sizeof(*nodetrack));
				if(!nodetrack)
					printfault(EXIT_NOMEM, "sim_worker_opexec_cnodes: Out of memory");
				hashtable_set(wo->node_tracking, sop->detail.cnodes.bootstrap_id, nodetrack);
				nodetrack->valid = 0;
				nodetrack->track_id = sop->detail.cnodes.bootstrap_id;
			}

			debugf("SWorker (%d): STALLING simop: Bootstrap attempt using unknown address, tracking ID %d\n", wo->process_id, nodetrack->track_id);

			/* return 0 == stall op */
			return 0;
		}
	}

	/* we'll setup all the nodes first, then instantiate them */
	list_tp instantiate_list = list_create();

	for(i=0; i<sop->detail.cnodes.quantity; i++) {
		in_addr_t addr;
		context_provider_tp context_provider;
		module_instance_tp modinst;

		/* every node contains its own context */
		context_provider = malloc(sizeof(*context_provider));

		if(!context_provider)
			printfault(EXIT_NOMEM, "sim_worker_opexec_cnodes: Out of memory");

		/* allocate module memory */
		modinst = context_provider->modinst = module_create_instance(module);

		/* assign an IP and start tracking */
		context_provider->address = addr = vci_create_ip(wo->vci_mgr, sop->detail.cnodes.network_id, context_provider);

		if(addr == INADDR_NONE) {
			dlogf(LOG_ERR, "SWorker: Failure to instantiate node!\n");
			return 1;
		}

		/* setup node logging channel */
		context_provider->log_channel = sop->detail.cnodes.logging_channel;
		context_provider->log_level = sop->detail.cnodes.logging_level;

		/* do we need to track this node?? */
		if(sop->detail.cnodes.flags & SIMOP_CNODES_TRACKING) {
			sim_worker_nodetracker_tp nodetrack;
			nbdf_tp nb_track;

			/* just in case... */
			sop->detail.cnodes.quantity = 1;

			nb_track = nbdf_construct("ia", sop->detail.cnodes.tracker_id, addr);
			dvn_packet_route(DVNPACKET_GLOBAL_BCAST, DVNPACKET_LAYER_SIM, 0, SIM_FRAME_TRACKNODE, nb_track);
			nbdf_free(nb_track);

			/* also track locally */
			nodetrack = malloc(sizeof(*nodetrack));
			if(!nodetrack)
				printfault(EXIT_NOMEM, "sim_worker_opexec_cnodes: Out of memory");

			nodetrack->track_id = sop->detail.cnodes.tracker_id;
			nodetrack->addr = addr;
			nodetrack->valid = 1;
			hashtable_set(wo->node_tracking, nodetrack->track_id, nodetrack);

			debugf( "SWorker (%d): Broadcast node tracking packet for %s on ID %d\n", wo->process_id, inet_ntoa_t(addr), sop->detail.cnodes.tracker_id);
		}

		list_push_back(instantiate_list, context_provider);
	}

	/* parse the cl_args into separate strings */
        list_tp args = list_create();
        char* result = strtok(sop->detail.cnodes.cl_args, " ");
        while(result != NULL) {
                list_push_back(args, result);
                result = strtok(NULL, " ");
        }

        /* setup for instantiation */
        int argc = list_get_size(args);
        char* argv[argc];
        int argi = 0;
        for(argi = 0; argi < argc; argi++) {
                argv[argi] = list_pop_front(args);
        }
        list_destroy(args);

        /* call module instantiation */
        while(list_get_size(instantiate_list) > 0) {
                context_provider_tp cp = list_pop_front(instantiate_list);
                context_execute_instantiate(cp, bootstrap_ip, argc, argv);
        }
        list_destroy(instantiate_list);

	return 1;
}

int sim_worker_opexec_end(sim_worker_tp wo, simop_tp sop) {
	nbdf_tp complete_frame;

	complete_frame = nbdf_construct("i", wo->process_id);
	dvn_packet_route(DVNPACKET_LOCAL_SLAVE, DVNPACKET_LAYER_SIM, 0, SIM_FRAME_DONE_WORKER, complete_frame);
	nbdf_free(complete_frame);

	wo->mode = sim_worker_mode_complete;
	debugf("SWorker: Simulation is complete.\n", wo->process_id);

	return 1;
}

int sim_worker_opexec_module_load(sim_worker_tp wo, simop_tp sop) {
	module_tp mod;

	debugf("SWorker (%d): Loading Module: %s\n", wo->process_id,sop->detail.modload.file);

	mod = module_load(wo->mod_mgr, sop->detail.modload.id, sop->detail.modload.file);

	if(mod)
		context_execute_init(mod);

	else {
		char buffer[200];

		snprintf(buffer,200,"Unable to load and validate '%s'", sop->detail.modload.file);
		sim_worker_abortsim(wo, buffer);
	}

	return 1;
}

int sim_worker_opexec_network(sim_worker_tp wo, simop_tp sop) {
	if(sop->detail.network.method == SIMOP_NETWORK_CREATE)
		vci_network_create(wo->vci_mgr, sop->detail.network.id1,
				sop->detail.network.base_delay, sop->detail.network.width, sop->detail.network.tail_width,
				sop->detail.network.reliability);

	else if(sop->detail.network.method == SIMOP_NETWORK_CONNECT)
		vci_network_connect(wo->vci_mgr, sop->detail.network.id1, sop->detail.network.id2,
				sop->detail.network.base_delay, sop->detail.network.width, sop->detail.network.tail_width,
				sop->detail.network.reliability);

	else if(sop->detail.network.method == SIMOP_NETWORK_DISCONNECT)
		vci_network_disconnect(wo->vci_mgr, sop->detail.network.id1, sop->detail.network.id2);

	return 1;
}

int sim_worker_opexec(sim_worker_tp wo, simop_tp sop) {
	int rv=1;

	switch(sop->opcode) {
		case SIMOP_CODE_CNODES:
			rv = sim_worker_opexec_cnodes(wo,sop);
			break;

		case SIMOP_CODE_END:
			rv = sim_worker_opexec_end(wo,sop);
			break;

		case SIMOP_CODE_MODLOAD:
			rv = sim_worker_opexec_module_load(wo, sop);
			break;

		case SIMOP_CODE_NETWORK:
			rv = sim_worker_opexec_network(wo, sop);
			break;
	}

	if(rv)
		free(sop);

	return rv;
}

void sim_worker_abortsim(sim_worker_tp wo, char * error) {
	nbdf_tp error_frame;

	error_frame = nbdf_construct("s", error);
	dvn_packet_route(DVNPACKET_MASTER, DVNPACKET_LAYER_SIM, 0, SIM_FRAME_ERROR, error_frame);
	nbdf_free(error_frame);

	/* set error mode flag */
	wo->mode = sim_worker_mode_error;

	debugf(LOG_ERR, "SWorker (%d): Aborting simulation. Error: %s\n", wo->process_id, error);
}

void sim_worker_destroy_node(sim_worker_tp wo, context_provider_tp cp) {
	/* free his dtimer endpoint */
	dtimer_destroy_timers(global_sim_context.sim_worker->timer_mgr, cp);

	/* free his ip! this will also free the associated module memory AND the context provider. */
	vci_free_ip(global_sim_context.sim_worker->vci_mgr, cp->address);

	/* free his globals and such */
	module_destroy_instance(cp->modinst);

	/* final free */
	free(cp);
}


void sim_worker_hashwalk_free(void * v, int k) {
	if(v)
		free(v);
}

void sim_worker_destroy(sim_worker_tp sim) {
	void * event;
	int event_type;

	vci_mgr_destroy(sim->vci_mgr);
	vci_destroy_addressing_scheme(sim->ascheme);
	dtimer_destroy_manager(sim->timer_mgr);
	module_mgr_destroy(sim->mod_mgr);

	while((event = events_dequeue(sim->events, NULL, &event_type)) != NULL) {
		switch(event_type){
			case EVENTS_TYPE_DTIMER:
				dtimer_free_event(event);
				break;
			case EVENTS_TYPE_VCI:
				vci_free_event(event);
				break;
			case EVENTS_TYPE_SIMOP:
				free(event);
				break;
		}
	}
	events_destroy(sim->events);

	while(list_get_size(sim->stalled_simops)){
		free(list_pop_front(sim->stalled_simops));
	}
	list_destroy(sim->stalled_simops);

	hashtable_walk(sim->node_tracking, &sim_worker_hashwalk_free);
	hashtable_destroy(sim->node_tracking);

	free(sim->worker_states);

	free(sim);
}


